// #include <iostream>
// #include <bits/stdc++.h>
// #include <sys/time.h>
// #include <time.h>
// #include <unistd.h>
// #include <chrono>
// #include <unistd.h>
// #include <fcntl.h>
// #include "../buttonrpc/buttonrpc.hpp"
// #include <mutex>
// #include <thread>
// #include <condition_variable>
// #include <atomic>
// #include <algorithm>

// using namespace std;
// #define COMMOM_PORT 1234
// #define HEART_BEART_PERIOD 100

// //通过传入raft.start()得到的返回值
// struct StartState{
//     int startIndex;
//     int startTerm;
//     bool isLeader;

//     StartState():startIndex(-1), startTerm(-1), isLeader(false){}
// };

// //一个存放当前raft的ID及自己两个RPC端口号的结构(为了减轻负担，一个选举，一个日志同步，分开来)
// struct PeersInfo{
//     pair<int, int> m_port;
//     int m_peerId;
// };

// //日志结构
// struct LogEntry{
//     string command;  // 命令
//     int    term;     // 任期

//     LogEntry(string cmd = "", int term = -1):command(cmd),term(term){}
// };

// // 持久化结构——在响应RPC之前，在稳定的存储上更新
// struct Persister{
//     int              currentTerm;  // 服务器看到的最新任期
//     int              votedFor;     // 当前任期内获得选票的候选人ID
//     vector<LogEntry> log;          // 日志条目；每个条目包含状态机的命令，以及领导者收到条目的时间
// };

// enum RAFT_STATE {LEADER = 0, CANDIDATE, FOLLOWER};          // 用枚举定义的raft三种状态

// // -------------------------------RPC参数-------------------------------------
// // RequestVote RPC 候选人为收集选票而调用
// // Arguments: 
// struct RequestVoteArgs{
//     int term;          // 候选人的任期号
//     int candidateId;   // 请求选票的候选人的 Id
//     int lastLogIndex;  // 候选人的最后日志条目的索引值
//     int lastLogTerm;   // 候选人最后日志条目的任期号
// };
// // Results: 
// struct RequestVoteReply{
//     int  term;         // 当前任期号，候选人会更新自己的任期号
//     bool VoteGranted;  // true 表示候选人获得了选票
// };

// // AppendEntries RPC 由领导者调用以复制日志条目；也作为心跳使用
// // Arguments: 
// struct AppendEntriesArgs{
//     int    term;          // 领导任期
//     int    leaderId;      // 使追随者可以为客户端重定向
//     int    prevLogIndex;  // 紧接在新日志之前的日志条目的索引
//     int    prevLogTerm;   // 紧邻新日志条目之前的日志条目的任期
//     int    leaderCommit;  // 领导者的已知已提交的最高的日志条目的索引
//     string sendLog;       // 需要被保存的日志条目（做心跳使用时，内容为空；为了提高效率可一次性发送多个）

//     friend Serializer& operator >> (Serializer& in, AppendEntriesArgs& d) {
// 		in >> d.term >> d.leaderId >> d.prevLogIndex >> d.prevLogTerm >> d.leaderCommit >> d.sendLog;
// 		return in;
// 	}
// 	friend Serializer& operator << (Serializer& out, AppendEntriesArgs d) {
// 		out << d.term << d.leaderId << d.prevLogIndex << d.prevLogTerm << d.leaderCommit << d.sendLog;
// 		return out;
// 	}
// };
// // Results: 
// struct AppendEntriesReply{
//     int  term;            // 当前任期，领导者会更新自己的任期
//     bool success;         // 如果跟随者所含有的条目和prevLogIndex以及prevLogTerm匹配，则为真
//     int  conflict_term;   // 用于冲突时日志快速匹配
//     int  conflict_index;  // 用于冲突时日志快速匹配                 
// };


// class Raft{
// public:
//     Raft() = default;
//     static void* listenForVote(void* arg);              // 用于监听voteRPC的server线程
//     static void* listenForAppend(void* arg);            // 用于监听appendRPC的server线程
//     static void* appendEntriesTicker(void* arg);        // 持续处理日志同步的守护线程
//     static void* electionTicker(void* arg);             // 持续处理选举的守护线程
//     static void* callRequestVote(void* arg, int id);    // 发voteRPC的线程
//     static void* sendAppendEntries(void* arg, int id);  // 发appendRPC的线程

//     void Make(vector<PeersInfo> peers, int id);                 // raft初始化
//     int getduration(timeval last);                              // 传入某个特定计算到当下的持续时间
//     void setBroadcastTime();                                    // 重新设定BroadcastTime，成为leader发心跳的时候需要重置
//     pair<int, bool> getState();                                 // 判断是否leader
//     RequestVoteReply requestVote(RequestVoteArgs args);         // vote的RPChandler
//     AppendEntriesReply appendEntries(AppendEntriesArgs args);   // append的RPChandler
//     bool checkLogUptodate(int term, int index);                 // 判断是否最新日志(两个准则)，vote时会用到
//     vector<LogEntry> getCmdAndTerm(string text);                // 用的RPC不支持传容器，所以封装成string，这个是解封装恢复函数
//     StartState start(string command);                           // 向raft传日志的函数，只有leader响应并立即返回，应用层用到
//     int getlastIndex();
//     int getlastTerm();
    
//     void serialize();      // 序列化
//     bool deserialize();    // 反序列化
//     void saveRaftState();  // 持久化
//     void readRaftState();  // 读取持久化状态
//     bool isKilled();       // ->check is killed?
//     void kill();           // 设定raft状态为dead
//     ~Raft() = default;
// private:
//     mutex r_lock;                 // 锁
//     condition_variable cond;      // 条件变量
//     vector<PeersInfo> peers;      // 投票信息
//     Persister persister;          // 持久化信息
//     int peerId;                   // 当前节点id
//     atomic_bool dead;

//     // 需要持久化的数据
//     int currentTerm; 
//     int votedFor;
//     vector<LogEntry> log;

//     // 所有服务器上的易失性状态
//     int commitIndex;  // 已知被提交的最高日志条目的索引（初始化为0，单调增加）
//     int lastApplied;  // 已应用于状态机的最高日志条目的索引（初始化为0，单调增加）

//     // 领导者（服务器）上的易失性状态
//     vector<int> nextIndex;   // 对于每个服务器，要发送给该服务器的下一个日志条目的索引（初始化为领导者的最后一个日志索引+1）
//     vector<int> matchIndex;  // 对于每个服务器，已知在服务器上复制的最高日志条目的索引（初始化为0，单调增加）

//     int recvVotes;     // 收到的票数
//     int finishedVote;  // 已                    投票的节点数量

//     RAFT_STATE state;
//     int leaderId;
//     struct timeval lastWakeTime;
//     struct timeval lastBroadcastTime;

// };

#include "raft.hpp"

void Raft::Make(vector<PeersInfo> peers, int id){
    this->peers = peers;
    this->peerId = id;
    this->dead = false;

    this->state = FOLLOWER;
    this->currentTerm = 0;
    this->leaderId = -1;
    this->votedFor = -1;
    gettimeofday(&this->lastWakeTime, NULL);

    this->recvVotes = 0;
    this->finishedVote = 0;

    this->lastApplied = 0;
    this->commitIndex = 0;
    this->nextIndex.resize(peers.size(), 1);
    this->matchIndex.resize(peers.size(), 0);

    thread listenForVote_tid(listenForVote, this);
    thread listenForAppend_tid(listenForAppend, this);
    listenForVote_tid.detach();
    listenForAppend_tid.detach();
}

// 获取间隔时间(ms)
int Raft::getduration(timeval last){
    struct timeval now;
    gettimeofday(&now, NULL);
    return ((now.tv_sec - last.tv_sec) * 1000000 + (now.tv_usec - last.tv_usec)) / 1000;
}

void Raft::setBroadcastTime(){
    gettimeofday(&lastBroadcastTime, NULL);
    if(lastBroadcastTime.tv_usec >= 200000){
        lastBroadcastTime.tv_usec -= 200000;
    }else{
        lastBroadcastTime.tv_sec -= 1;
        lastBroadcastTime.tv_usec += (1000000 - 200000);
    }
}

pair<int, bool> Raft::getState(){
    pair<int, bool> serverState;
    unique_lock<mutex> lock(r_lock);

    serverState.first = currentTerm;
    serverState.second = (state == LEADER);
    lock.unlock();
    return serverState;
}

void* Raft::listenForVote(void* arg){
    Raft* raft = (Raft*)arg;
    buttonrpc server;
    server.as_server(raft->peers[raft->peerId].m_port.first);
    server.bind("appendEntries", &Raft::requestVote, raft); // 监听AppendEntries RPC信息

    thread listenVote(electionTicker, raft);
    listenVote.detach();

    server.run();
}

void* Raft::listenForAppend(void* arg){
    Raft* raft = (Raft*)arg;
    buttonrpc server;
    server.as_server(raft->peers[raft->peerId].m_port.second);
    server.bind("appendEntries", &Raft::appendEntries, raft);

    thread listenAppend(appendEntriesTicker, raft);
    listenAppend.detach();

    server.run();
}

void* Raft::electionTicker(void* arg){
    Raft* raft = (Raft*)arg;
    bool resetFlag = false;
    while (!raft->dead.load()){
        int timeOut = rand()%200 + 200; // 心跳超时时间
        while (1){
            usleep(1000);
            unique_lock<mutex> lock(raft->r_lock);

            int duration = raft->getduration(raft->lastWakeTime);
            if (duration > timeOut && raft->state != LEADER ){
                cout << "Raft node:" << raft->peerId << "start election at term " << raft->currentTerm;
                // 转变状态
                raft->state = CANDIDATE;
                raft->currentTerm++;
                raft->votedFor = raft->peerId;
                raft->saveRaftState();

                raft->recvVotes = 1;
                raft->finishedVote = 1;

                // 开启多线程给每个raft节点发起投票请求
                for (int i = 0; i < raft->peers.size(); i++){
                    if (raft->peers[i].m_peerId == raft->peerId) continue;
                    thread tid(callRequestVote, raft);
                    tid.detach();
                }

                while(raft->recvVotes <= raft->peers.size() / 2 && raft->finishedVote != raft->peers.size()){
                    raft->cond.wait(lock);
                }
                
                if(raft->state != CANDIDATE){
                    raft->r_lock.unlock();
                    continue;
                }

                if(raft->recvVotes > raft->peers.size() / 2){
                    raft->state = LEADER;
                    for(int i = 0; i < raft->peers.size(); i++){
                        raft->nextIndex[i] = raft->log.size() + 1;
                        raft->matchIndex[i] = 0;
                    }

                    cout << raft->peerId << " become new leader at term " << raft->currentTerm << endl;
                    raft->setBroadcastTime();
                }
            }
            lock.unlock();
            if (resetFlag){
                resetFlag = false;
                break;
            }
        }  
    }
}

void* Raft::appendEntriesTicker(void* arg){
    Raft* raft = (Raft*) arg;
    while (!raft->dead.load()){
        usleep(1000);
        unique_lock<mutex> lock(raft->r_lock);
        if(raft->state != LEADER){
            lock.unlock();
            continue;
        }

        int duration = raft->getduration(raft->lastBroadcastTime);
        if (duration < HEART_BEART_PERIOD){
            lock.unlock();
            continue;
        }

        gettimeofday(&raft->lastBroadcastTime, NULL);
        lock.unlock();
        // 开启多线程给每个raft节点发起appendEntries请求
        for (int i = 0; i < raft->peers.size(); i++){
            if (raft->peers[i].m_peerId == raft->peerId) continue;
            thread tid(sendAppendEntries, raft);
            tid.detach();
        }
    }   
}

void* Raft::callRequestVote(void* arg, int id){
    Raft* raft = (Raft*) arg;
    buttonrpc client;     
    unique_lock<mutex> lock(raft->r_lock);

    RequestVoteArgs args;
    args.candidateId = raft->peerId;
    args.term = raft->currentTerm;
    args.lastLogIndex = raft->getlastIndex();
    args.lastLogTerm = raft->getlastTerm();

    client.as_client("127.0.0.1", raft->peers[id].m_port.first);
    lock.unlock();

    RequestVoteReply reply = client.call<RequestVoteReply>("requestVote", args).val();

    lock.lock();
    raft->finishedVote++;
    raft->cond.notify_one();
    // raft任期小于当前节点，立即转为FOLLOWER
    if (reply.term > raft->currentTerm){
        raft->state = FOLLOWER;
        raft->currentTerm = reply.term;
        raft->votedFor = -1;
        raft->readRaftState();
        lock.unlock();
        return nullptr;
    }
    if(reply.VoteGranted){
        raft->recvVotes++;
    }
    lock.unlock();
}

RequestVoteReply Raft::requestVote(RequestVoteArgs args){
    RequestVoteReply reply;
    reply.VoteGranted = false;

    unique_lock<mutex> lock(r_lock);
    reply.term = currentTerm;

    // 请求投票的CANDIDATE任期比当前节点小，不予投票
    if (currentTerm > args.term){
        lock.unlock();
        return reply;
    }
    // 请求投票的CANDIDATE任期比当前节点大，当前节点需置为FOLLOWER并同步任期
    if (currentTerm < args.term){
        state = FOLLOWER;
        currentTerm = args.term;
        votedFor = -1;
    }
    // 如果 votedFor 是 -1 或 candidateId，并且候选人的日志至少与接收人的日志一样新，则投票
    if(votedFor == -1 || votedFor == args.candidateId){
        lock.unlock();
        bool ret = checkLogUptodate(args.lastLogTerm, args.lastLogIndex); // 日志检查
        if (!ret) return reply;

        lock.lock();
        votedFor = args.candidateId;
        reply.VoteGranted = true;
        gettimeofday(&lastWakeTime, NULL);
    }
    saveRaftState();
    lock.unlock();
    return reply;
}

void* Raft::sendAppendEntries(void* arg, int id){
    Raft* raft = (Raft*)arg;
    buttonrpc client;
    AppendEntriesArgs args;
    unique_lock<mutex> lock(raft->r_lock);

    client.as_client("127.0.0.1", raft->peers[id].m_port.first);
    lock.unlock();

    args.term = raft->currentTerm;
    args.leaderId = raft->peerId;
    args.prevLogIndex = raft->nextIndex[id] - 1;
    args.leaderCommit = raft->commitIndex;

    for (int i = args.prevLogIndex; i < raft->log.size(); i++){
        args.sendLog += (raft->log[i].command + "," + to_string(raft->log[i].term) + ";");
    }
    if(args.prevLogIndex == 0){
        args.prevLogTerm = 0;
        if(raft->log.size() != 0){
            args.prevLogTerm = raft->log[0].term;
        }
    }
    else args.prevLogTerm = raft->log[args.prevLogIndex - 1].term;
    lock.unlock();

    AppendEntriesReply reply = client.call<AppendEntriesReply>("appendEntries", args).val();

    lock.lock();
    if (reply.term > raft->currentTerm){
        raft->state = FOLLOWER;
        raft->currentTerm = reply.term;
        raft->votedFor = -1;
        raft->saveRaftState();
        lock.unlock();
        return nullptr;
    }
    
    if (reply.success){  // 日志匹配
        raft->nextIndex[id] += raft->getCmdAndTerm(args.sendLog).size();
        raft->matchIndex[id] = raft->nextIndex[id] - 1;

        // 同一任期下大多数服务器上已匹配的日志索引大于raft已提交的最大日志索引,则commitIndex 更新为大多数服务器上已匹配的日志索引
        vector<int> tmpIndex = raft->matchIndex;
        sort(tmpIndex.begin(), tmpIndex.end());
        int realMajorityMatchIndex = tmpIndex[tmpIndex.size() / 2];
        if(realMajorityMatchIndex > raft->commitIndex && raft->log[realMajorityMatchIndex - 1].term == raft->currentTerm){
            raft->commitIndex = realMajorityMatchIndex;
        }
    } else{   // 跟随者所含有的条目和prevLogIndex以及prevLogTerm不匹配
        if (reply.conflict_term != -1){  // term冲突
            int leader_conflict_index = -1;
            for (int i = args.prevLogIndex; i >= 1; i--){
                if (raft->log[i - 1].term == reply.conflict_term){
                    leader_conflict_index = i;
                    break;
                }
            }
            if (leader_conflict_index != -1){
                raft->nextIndex[id] = leader_conflict_index + 1;
            } else {
                raft->nextIndex[id] = reply.conflict_index;
            }
        } else {   // index冲突
            raft->nextIndex[id] = reply.conflict_index + 1;
        }
    }
    raft->saveRaftState();
    lock.unlock();
}

AppendEntriesReply Raft::appendEntries(AppendEntriesArgs args){
    vector<LogEntry> recvLog = getCmdAndTerm(args.sendLog);
    AppendEntriesReply reply;
    unique_lock<mutex> lock(r_lock);
    reply.term = currentTerm;
    reply.success = false;
    reply.conflict_index = -1;
    reply.conflict_term = -1;
    
    if (args.term < currentTerm){ // 发送心跳的leader任期比当前节点小
        lock.unlock();
        return reply;
    } else if (args.term > currentTerm){  // 发送心跳的leader任期比当前节点大
        votedFor = -1;
        saveRaftState();
        currentTerm = args.term;
        state = FOLLOWER;
    } else {
        state = FOLLOWER;
    }
    gettimeofday(&lastWakeTime, NULL);

    int logSize = 0;
    if(log.size() == 0){
        for(const auto& l : recvLog){
            log.emplace_back(l);
        }
        saveRaftState();
        logSize = log.size();
        if(commitIndex < args.leaderCommit){
            commitIndex = min(args.leaderCommit, logSize);
        }
        lock.unlock();
        reply.success = true;
        return reply;
    }

    if (log.size() < args.prevLogIndex){
        reply.conflict_index = log.size();
        lock.unlock();
        return reply;
    }
    if (args.prevLogIndex > 0 && log[args.prevLogIndex - 1].term != args.prevLogTerm){
        reply.conflict_term = log[args.prevLogIndex - 1].term;
        for (int i = args.prevLogIndex; i >= 1; i--){
            if (log[i - 1].term != reply.conflict_term){
                reply.conflict_index = i + 1;
                break;
            }            
        }
        lock.unlock();
        return reply;
    }

    logSize = log.size();
    for(int i = args.prevLogIndex; i < logSize; i++){
        log.pop_back();
    }
    for(const auto& l : recvLog){
        log.emplace_back(l);
    }
    saveRaftState();

    if (args.leaderCommit > commitIndex){
        commitIndex = min(args.leaderCommit, static_cast<int>(log.size()));
    }
    cout << "raft " << peerId << " sync success!" << endl;
    lock.unlock();
    reply.success = true;
    return reply;
}

void* Raft::applyLogLoop(void* arg){
    Raft* raft = (Raft*)arg;
    unique_lock<mutex> lock(raft->r_lock);
    lock.unlock();
    while (1){
        while(!raft->dead.load()){
            usleep(10000);
            vector<ApplyMsg> msgs;
            lock.lock();
            while(raft->lastApplied < raft->commitIndex){
                raft->lastApplied++;
                ApplyMsg msg;
                msg.command = raft->log[raft->lastApplied].command;
                msg.commandValid = true;
                msg.commandTerm = raft->log[raft->lastApplied].term;
                msg.commandIndex = raft->lastApplied;
                msgs.emplace_back(msg);
            }
            lock.unlock();
            // 与KVServer交互执行command
            for (int i = 0; i < msgs.size(); i++){
                raft->MQ.Push(msgs[i]);
            }
        }
        usleep(10000);
    }
}

StartState Raft::start(string command){
    StartState ret;
    unique_lock<mutex> lock(r_lock);

    RAFT_STATE state = state;
    if(state != LEADER){
        lock.unlock();
        return ret;
    }
    ret.startIndex = log.size();
    ret.startTerm = currentTerm;
    ret.isLeader = true;

    LogEntry rlog;
    rlog.command = command;
    rlog.term = currentTerm;
    log.emplace_back(rlog);
    lock.unlock();

    return ret;
}

bool Raft::checkLogUptodate(int term, int index){
    unique_lock<mutex> lock(r_lock);
    int lastIndex = getlastIndex();
    int lastTerm = getlastTerm();
    if (term > lastTerm || term == lastTerm && index >= lastIndex){
        lock.unlock();
        return true;
    }
    lock.unlock();
    return false;
}

void Raft::saveRaftState(){
    persister.currentTerm = currentTerm;
    persister.votedFor = votedFor;
    persister.log = log;
    serialize();
}

void Raft::readRaftState(){
    //只在节点启动的初始化时候调用，没必要加锁，因为run()在其之后才执行
    bool ret = this->deserialize();
    if(!ret) return;
    this->currentTerm = this->persister.currentTerm;
    this->votedFor = this->persister.votedFor;

    for(const auto& l : this->persister.log){
        this->log.emplace_back(l);
    }
}

bool Raft::deserialize() {
    std::string filename = "persister-" + std::to_string(peerId);
    std::ifstream file(filename); // 打开文件以读取内容

    if (!file) {
        return false; // 文件不存在，反序列化失败
    }

    std::string content((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
    file.close();

    std::vector<std::string> persist;
    std::string tmp = "";
    for (int i = 0; i < content.size(); i++) {
        if (content[i] != ';') {
        tmp += content[i];
        } else {
        if (tmp.size() != 0) persist.push_back(tmp);
        tmp = "";
        }
    }
    persist.push_back(tmp);

    this->persister.currentTerm = std::stoi(persist[0]);
    this->persister.votedFor = std::stoi(persist[1]);

    std::vector<std::string> log;
    std::vector<LogEntry> logs;
    tmp = "";
    for (int i = 0; i < persist[2].size(); i++) {
        if (persist[2][i] != '.') {
        tmp += persist[2][i];
        } else {
        if (tmp.size() != 0) log.emplace_back(tmp);
        tmp = "";
        }
    }

    for (int i = 0; i < log.size(); i++) {
        tmp = "";
        int j = 0;
        for (; j < log[i].size(); j++) {
        if (log[i][j] != ',') {
            tmp += log[i][j];
        } else {
            break;
        }
        }
        std::string number(log[i].begin() + j + 1, log[i].end());
        int num = std::stoi(number);
        logs.emplace_back(LogEntry(tmp, num));
    }

    this->persister.log = logs;
    return true;
}

vector<LogEntry> Raft::getCmdAndTerm(string text){
    vector<LogEntry> logs;
    int n = text.size();
    vector<string> str;
    string tmp = "";
    for(int i = 0; i < n; i++){
        if(text[i] != ';'){
            tmp += text[i];
        }else{
            if(tmp.size() != 0) str.push_back(tmp);
            tmp = "";
        }
    }
    for(int i = 0; i < str.size(); i++){
        tmp = "";
        int j = 0;
        for(; j < str[i].size(); j++){
            if(str[i][j] != ','){
                tmp += str[i][j];
            }else break;
        }
        string number(str[i].begin() + j + 1, str[i].end());
        int num = atoi(number.c_str());
        logs.push_back(LogEntry(tmp, num));
    }
    return logs;
}

void Raft::serialize() {
  std::string filename = "persister-" + std::to_string(peerId);
  std::ofstream file(filename, ios_base::out | ios_base:: app); // 打开文件以写入内容

  if (!file) {
    std::cerr << "Failed to open file: " << filename << std::endl;
    exit(-1);
  }

  file << this->persister.currentTerm << ";" << this->persister.votedFor << ";";

  for (const auto& log : this->persister.log) {
    file << log.command << "," << log.term << ".";
  }

  file.close();
}

bool Raft::isKilled(){
    return this->dead.load();
}

void Raft::kill(){
    this->dead.store(true);
}

void Raft::activate(){
    dead = 0;
    printf("raft%d is activate\n", peerId);
}

int Raft::getlastIndex(){
    return log.size();
}

int Raft::getlastTerm(){
    int lastTerm;
    if(log.size() != 0){
        lastTerm = log.back().term;
    }
    return lastTerm;
}

int main(int argc, char* argv[]){
    if(argc < 2){
        cout << "loss parameter of peersNum" << endl;
        exit(-1);
    }

    int peersNum = atoi(argv[1]);
    if(peersNum % 2 == 0){
        cout << "the peersNum should be odd" << endl;  //必须传入奇数，这是raft集群的要求
        exit(-1);
    }

    srand((unsigned)time(NULL));
    vector<PeersInfo> peers(peersNum);
    for(int i = 0; i < peersNum; i++){
        peers[i].m_peerId = i;
        peers[i].m_port.first = COMMOM_PORT + i;                    //vote的RPC端口
        peers[i].m_port.second = COMMOM_PORT + i + peers.size();    //append的RPC端口
    }
    Raft* raft = new Raft[peers.size()];
    for(int i = 0; i < peers.size(); i++){
        raft[i].Make(peers, i);
    }
}